package com.cmcc.regionmanager;

import com.google.common.base.Strings;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class DeleteRegion {
    private static final String HBASEROOTDIR="hbase.rootdir";
    private static Configuration conf = HBaseConfiguration.create();

    public static void main(String[] args) throws IOException {
        if(args.length<2){
            System.out.println("please set tablename startkey");
            System.exit(-1);
        }
        String tablename = args[0];
        String startkey = args[1];

        Connection connection = ConnectionFactory.createConnection(conf);
        Admin admin = connection.getAdmin();

        FileSystem fs=FileSystem.get(conf);
        TableName srctableName = TableName.valueOf(tablename);
        Table metatable = connection.getTable(TableName.META_TABLE_NAME);


        List<HRegionInfo> regionInfoList = admin.getTableRegions(srctableName);


        for(HRegionInfo regionInfo: regionInfoList){
            String regionStartKey= new String(regionInfo.getStartKey());
            if(startkey.equals(regionStartKey)){

                System.out.println(regionInfo.getRegionNameAsString());


                closeRegion(admin,metatable,regionInfo);
                admin.disableTable(srctableName);
                deleteRegionData(fs,regionInfo);

                deleteRegionMeta(metatable,regionInfo);
                admin.enableTable(srctableName);
            }
        }
        admin.close();
        connection.close();

    }

    static String getHbaseRegionPath(HRegionInfo regionInfo) throws IOException {
        String hbaseroot =  conf.get(HBASEROOTDIR);
        if(Strings.isNullOrEmpty(hbaseroot)) {
            throw new IOException("parameter hbase.rootdir not set");
        }
        TableName tableName = regionInfo.getTable();
        String regionpath=hbaseroot+"/data/"+
                tableName.getNameAsString().replace(":","/")+
                "/"+regionInfo.getEncodedName();
        return regionpath;
    }

    static  void deleteRegionData(FileSystem fs , HRegionInfo  regionInfo) throws IOException {
        String regionpath = getHbaseRegionPath(regionInfo);
        System.out.println(regionpath);
        fs.delete(new Path(regionpath),true);
    }

    static void closeRegion(Admin admin,Table metatable,HRegionInfo regionInfo) throws IOException {
        Get get = new Get(regionInfo.getRegionName());
        Result  rs= metatable.get(get);
        if(rs.isEmpty()){
            System.out.println("not exist this region , please check it");
            System.exit(-1);
        }
        String regionserverName = new String(rs.getValue(Bytes.toBytes("info"),Bytes.toBytes("server")));
        long  regionserverCode = rs.getValueAsByteBuffer(Bytes.toBytes("info"),Bytes.toBytes("serverstartcode")).getLong();
        ServerName serverName =  ServerName.valueOf(regionserverName,regionserverCode);
        admin.closeRegion(serverName,regionInfo);
    }

    static void deleteRegionMeta(Table metatbl, HRegionInfo  regionInfo) throws IOException {
        List list = new ArrayList<Delete>();
        Delete delete = new Delete( regionInfo.getRegionNameAsString().getBytes());
        list.add(delete);
        metatbl.delete(delete);
    }

}
